package com.bd08.flink.demo






import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.assigners.{TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time


object WindowTWordCount  extends App {

  case class WordCount(word:String,count:Int)


  val env = StreamExecutionEnvironment.getExecutionEnvironment
  //val ds = env.addSource(new SocketTextStreamFunction("172.18.172.150",9999,"\n",3))
  val ds = env.socketTextStream("172.18.169.160",9999,'\n')
    //.assignAscendingTimestamps(line => line.hashCode)

    .flatMap(line => line.split("\\s")).map(WordCount(_,1)).keyBy(wc => wc.word)

    //.window(TumblingProcessingTimeWindows.of(Time.seconds(30))).sum("count")
    .countWindow(5).sum("count")

  ds.print()
  env.execute("window-t-word-count")


}
